package com.shujia.state

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable

object Demo01NoStateWordCount {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(8)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * 统计每个单词的数量
     */
    linesDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .keyBy(_._1)
      /**
       * process转换：每接收一条数据 会处理一次（调用processElement方法），类似map方法
       */
      .process(new KeyedProcessFunction[String, (String, Int), (String, Int)] {
        // 每一个并行度会创建一次，一个并行度中的所有Key会共用一个变量
        //        var cnt: Int = 0
        // 使用HashMap来区分一个并行度中的不同的Key，保存每个数量单词的
        /**
         * Flink的状态：某一时刻的计算结果
         * 如果直接用Map保存Flink的状态，随着任务的不断运行
         * 会导致内存占用越来越大，最终可能导致程序崩溃
         * 而且如果直接将状态保存到内存中是非常不可靠的
         */
        val wordMap = new mutable.HashMap[String, Int]()

        /**
         *
         * @param value 当前处理的一条数据
         * @param ctx   上下文对象
         * @param out   用于将数据发送到下游任务
         */
        override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {
          val word: String = value._1
          val cnt: Int = wordMap.getOrElse(word, 0)
          val newCnt: Int = cnt + 1
          wordMap.put(word, newCnt)
          out.collect((value._1, newCnt))
        }
      })
      .print()

    env.execute()


  }

}
